package executors

import (
	"fmt"
	"log"
	"math"
	"sync/atomic"
	"time"
)

var ErrClosedExecutors = fmt.Errorf("executors is closed")
var ErrClosingExecutors = fmt.Errorf("executors is closing")
var ErrWriteChannelFailed = fmt.Errorf("write task channel failed, Probably the task channel is full!")

const (
	running int32 = iota
	closing
	closed
)

type Pool struct {
	handler func(interface{})
	recover func(error)
	taskCh  chan interface{}
	limit   int32
	count   int32
	running int32
	status  int32
}

//New 创建一个新的协程池
func New(limit int, handler func(interface{}), recovers ...func(error)) *Pool {
	if limit >= math.MaxInt32 {
		log.Panicf("最大协程数不得超过%d\n", math.MaxInt32)
	}
	p := &Pool{
		handler: handler,
		recover: nil,
		taskCh:  nil,
		limit:   int32(limit),
		count:   0,
		running: 0,
		status:  0,
	}
	if len(recovers) > 0 {
		p.recover = recovers[0]
	} else {
		p.recover = func(err error) {
			log.Println(err)
		}
	}

	p.taskCh = make(chan interface{}, 1024)

	for i := 0; i < limit; i++ {
		p.fork()
	}

	go p.monitor()

	return p
}

func (p *Pool) AddTask(job interface{}) error {
	if p.status == closing {
		return ErrClosingExecutors
	}
	if p.status == closed {
		return ErrClosedExecutors
	}

	select {
	case p.taskCh <- job:
		return nil
	case <-time.After(time.Second):
		return ErrWriteChannelFailed
	}
	return nil
}

func (p *Pool) Close() {
	if p.status == running {
		p.status = closing
	}
}

func (p *Pool) WaitClose() {
	p.Close()
	t := time.NewTicker(time.Millisecond * 200)
	for {
		<-t.C
		if p.status == closed {
			return
		}
	}
}

func (p *Pool) monitor() {
	t := time.NewTicker(time.Second)
	for {
		<-t.C
		if p.status == closed {
			return
		}

		if p.count < p.limit {
			diff := p.limit - p.count
			if p.status == running {
				var n int32 = 0
				for ; n < diff; n++ {
					p.fork()
				}
			}
			if p.status == closing {
				if len(p.taskCh) > 0 {
					var n int32 = 0
					for ; n < diff; n++ {
						p.fork()
					}
				}
			}
		}
	}
}

func (p *Pool) fork() {
	atomic.AddInt32(&p.count, 1)
	go func() {
		defer func() {
			atomic.AddInt32(&p.count, -1)
		}()
		defer func() {
			if err := recover(); err != nil {
				p.recover(fmt.Errorf("%v", err))
			}
		}()

		ticker := time.NewTicker(time.Second * 1)
		for {
			select {
			case t := <-p.taskCh:
				p.handler(t)

			case <-ticker.C:
				if p.status == closing {
					if len(p.taskCh) == 0 {
						p.status = closed
						return
					}
				} else if p.status == closed {
					return
				}
			}
		}
	}()
}
